package cn.spdb.harrier.rpc.remote;

import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.spdb.harrier.common.utils.NameThreadFactory;
import cn.spdb.harrier.common.utils.URI;
import cn.spdb.harrier.rpc.client.ConsumerConfig;
import cn.spdb.harrier.rpc.client.RpcFuture;
import cn.spdb.harrier.rpc.client.RpcRequestCache;
import cn.spdb.harrier.rpc.client.RpcRequestTable;
import cn.spdb.harrier.rpc.code.NettyDecoder;
import cn.spdb.harrier.rpc.code.NettyEncoder;
import cn.spdb.harrier.rpc.common.RpcRequest;
import cn.spdb.harrier.rpc.common.RpcResponse;
import cn.spdb.harrier.rpc.config.NettyClientConfig;
import cn.spdb.harrier.rpc.protocol.RpcProtocol;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;

public class NettyClient {

	private static class NettyClientInner {
		private static final NettyClient instance = new NettyClient(new NettyClientConfig());
	}

	public static NettyClient getInstance() {
		return NettyClient.NettyClientInner.instance;
	}

	private final Logger logger = LoggerFactory.getLogger(NettyClient.class);

	private final EventLoopGroup workGroup;
	private final NettyClientConfig nettyClientConfig;
	private final Class<? extends SocketChannel> socketChannelClass;
	private final Bootstrap bootstrap = new Bootstrap();

	private final AtomicBoolean isStarted = new AtomicBoolean(false);

	private final ConcurrentHashMap<URI, Channel> channelMap = new ConcurrentHashMap<URI, Channel>();

	private int outTime = 5 * 60 * 1000;

	public Channel getChannel(URI uri) {
		Channel channel = channelMap.get(uri);
		if (channel != null && channel.isActive()) {
			return channel;
		}
		return createChannel(uri, true);
	}

/**
 * 创建与指定 URI 地址的 Netty 客户端连接通道
 *
 * @param uri   远程服务地址信息，包含 IP 和端口
 * @param isSync 是否以同步方式建立连接
 * @return 成功连接则返回 Channel 对象，否则返回 null
 */
private Channel createChannel(URI uri, boolean isSync) {
    ChannelFuture channelFuture;
    try {
        // 使用 bootstrap 建立远程连接，防止并发连接创建冲突，使用 synchronized 保证线程安全
        synchronized (bootstrap) {
            channelFuture = bootstrap.connect(new InetSocketAddress(uri.getIp(), uri.getPort()));

            // 如果是同步连接，则等待连接完成
            if (isSync) {
                channelFuture.sync();
            }

            // 判断连接是否成功
            if (channelFuture.isSuccess()) {
                Channel channel = channelFuture.channel();

                // 将成功建立的通道缓存起来，避免重复连接
                channelMap.put(uri, channel);

                // 返回连接通道
                return channel;
            }
        }
    } catch (Exception e) {
        // 捕获并记录连接异常日志
        logger.warn(String.format("connect to %s error", uri), e);
    }

    // 连接失败返回 null
    return null;
}


	private NettyClient(final NettyClientConfig clientConfig) {
		this.nettyClientConfig = clientConfig;
		this.workGroup = NettyEventLoopFactory.eventLoopGroup(nettyClientConfig.getWorkerThreads(),
				"NettyClientWorkThread");
		this.socketChannelClass=NettyEventLoopFactory.socketChannelClass();
		this.start();
	}

/**
 * 初始化并启动 Netty 客户端。
 * <p>
 * 该方法配置客户端的连接参数、网络协议编解码器、心跳机制等，
 * 并启动后台线程用于管理连接超时和清理无效 Channel。
 */
private void start() {
    // 配置 Bootstrap，设置事件循环组、通道类型及各种连接选项
    this.bootstrap.group(workGroup)
                  .channel(this.socketChannelClass) // 指定客户端 SocketChannel 实现类
                  .option(ChannelOption.SO_KEEPALIVE, nettyClientConfig.isSoKeepalive()) // 是否启用 TCP KeepAlive
                  .option(ChannelOption.TCP_NODELAY, nettyClientConfig.isTcpNoDelay())   // 是否禁用 Nagle 算法
                  .option(ChannelOption.SO_SNDBUF, nettyClientConfig.getSendBufferSize()) // 设置发送缓冲区大小
                  .option(ChannelOption.SO_RCVBUF, nettyClientConfig.getReceiveBufferSize()); // 设置接收缓冲区大小

    // 添加日志处理器和自定义 Channel 初始化处理器
    this.bootstrap.handler(new LoggingHandler(LogLevel.DEBUG)) // 记录通信日志（调试级别）
                      .handler(new ChannelInitializer<SocketChannel>() {

                          /**
                           * 初始化每个新建立的 SocketChannel 的处理管道（pipeline）
                           */
                          @Override
                          protected void initChannel(SocketChannel ch) throws Exception {
                              ch.pipeline()
                                .addLast("encoder", new NettyEncoder()) // 请求编码器（将对象转为字节流）
                                .addLast("decoder", new NettyDecoder(RpcResponse.class)) // 响应解码器（将字节流转为对象）
                                .addLast("client-idle-handler", new IdleStateHandler(6 * 1000, 0, 0, TimeUnit.MILLISECONDS)) // 客户端空闲检测
                                .addLast("handler", new NettyClientHandler()); // 自定义业务处理器
                          }
                      });

    // 标记客户端已启动
    isStarted.compareAndSet(false, true);

    // 创建一个守护线程，定期检查并清理失效的连接
    NameThreadFactory factory = new NameThreadFactory(this.getClass().getSimpleName() + "_out_time_channel");
    factory.newDeamonThread(() -> {
        while (isStarted.get()) {
            try {
                // 每隔 outTime 时间（默认5分钟）执行一次清理
                Thread.sleep(outTime);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            // 遍历所有缓存的 Channel，移除不活跃的连接
            channelMap.entrySet().iterator().forEachRemaining(action -> {
                if (!action.getValue().isActive()) {
                    channelMap.remove(action.getKey());
                }
            });
        }
    }).start();
}


/**
 * 向服务端发送 RPC 协议消息（请求），并根据配置决定是否异步等待响应结果。
 *
 * @param uri             服务端地址标识
 * @param protocol        封装了RPC请求的协议对象
 * @param consumerConfig  消费者配置，包含调用方式（同步/异步）、超时时间等
 * @return 返回 RPC 调用的响应结果（同步调用时会阻塞等待，异步则立即返回默认响应）
 */
public RpcResponse sendMsg(final URI uri, RpcProtocol<RpcRequest> protocol, ConsumerConfig consumerConfig) {
    // 获取与目标服务器的连接通道
    Channel channel = getChannel(uri);
    assert channel != null;

    // 提取请求体和请求ID
    RpcRequest request = protocol.getBody();
    RpcRequestCache rpcRequestCache = new RpcRequestCache();
    String handlerMethName = request.getHandlerMethName(); // 获取处理方法名称，用于日志和缓存
    rpcRequestCache.setHandlerMethName(handlerMethName);

    Long reqId = protocol.getProtocolHeader().getRequestId(); // 请求唯一标识
    RpcFuture future = null;

    // 如果是同步调用，则创建一个 Future 来等待响应结果
    if (Boolean.FALSE.equals(consumerConfig.getAsync())) {
        future = new RpcFuture(request, reqId);
        rpcRequestCache.setFuture(future);
    }

    // 将本次请求放入全局请求表中，用于后续接收响应时匹配处理结果
    RpcRequestTable.put(protocol.getProtocolHeader().getRequestId(), rpcRequestCache);

    // 发送协议数据到远程服务端
    channel.writeAndFlush(protocol);

    RpcResponse rpcResponse = null;

    // 如果是异步调用，直接返回一个成功状态的空响应
    if (Boolean.TRUE.equals(consumerConfig.getAsync())) {
        rpcResponse = new RpcResponse();
        rpcResponse.setStatus((byte) 0); // 设置状态为成功
        rpcResponse.setResult(true);    // 表示已成功发送
        return rpcResponse;
    }

    // 否则为同步调用，等待响应或超时
    try {
        assert future != null;
        rpcResponse = future.get(consumerConfig.getTimeOut(), TimeUnit.MILLISECONDS);
    } catch (Exception e) {
        // 记录发送失败的日志
        logger.error("send msg error service name is {}", handlerMethName, e);
    }

    return rpcResponse;
}


/**
 * 关闭 Netty 客户端，释放所有资源。
 * <p>
 * 该方法保证只执行一次，通过 CAS 操作确保线程安全。
 * 主要操作包括：
 * - 关闭所有已建立的连接通道（Channel）
 * - 关闭并释放底层的 EventLoopGroup 线程池资源
 */
public void close() {
    // 使用 CAS 判断是否已经关闭，避免重复关闭
    if (isStarted.compareAndSet(true, false)) {
        try {
            // 关闭所有与服务端建立的连接通道
            closeChannels();

            // 关闭事件循环组（即底层的线程池）
            if (workGroup != null) {
                this.workGroup.shutdownGracefully();
            }
        } catch (Exception e) {
            // 捕获并记录关闭过程中发生的异常
            logger.error("netty client close exception", e);
        }

        // 记录客户端关闭日志
        logger.info("netty client closed");
    }
}

/**
 * 关闭所有缓存中的 Channel 连接，并清空缓存
 */
private void closeChannels() {
    // 遍历所有保存的 Channel 并关闭
    for (Channel channel : this.channelMap.values()) {
        channel.close();
    }

    // 清空 Channel 缓存映射表
    this.channelMap.clear();
}


}
